package nio.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;
import java.util.Set;

/**
 * @author Huhailong
 * @Description 多路复用类 负责轮询多路复用器Selector,可以处理多个客户端的并发接入
 * @Date 2021/4/9.
 */
public class MultiplexerTimeServer implements Runnable {

    private Selector selector;

    private volatile boolean stop;

    /**
     * 初始化多路复用器，绑定监听端口
     *
     * @param port 绑定端口
     */
    public MultiplexerTimeServer(int port) {
        try {
            selector = Selector.open();
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false); //配置为非阻塞
            serverChannel.socket().bind(new InetSocketAddress(port), 1024);
            System.out.println("服务器地址：" + new InetSocketAddress("127.0.0.1", port).getHostString());
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);   //注册selector并监听ACCEPT操作
            System.out.println("Time Server is start in port: " + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    public void stop() {
        this.stop = true;
    }

    @Override
    public void run() {
        while (!stop) {
            try {
                selector.select(1000);  //超时设置
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                SelectionKey key;
                while (iterator.hasNext()) {
                    key = iterator.next();
                    iterator.remove();
                    handleInput(key);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 处理请求接入
     *
     * @param key selection key
     */
    private void handleInput(SelectionKey key) throws IOException {
        /*
        判断key是否有效
        一个键在创建时是有效的，直到它被取消、它的通道关闭或它的选择器关闭为止
         */
        try {
            if (key.isValid()) {
                System.out.println("有效的key进入");
                //处理新接入的请求消息
                if (key.isAcceptable()) {
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    System.out.println(socketChannel.getRemoteAddress()+" join system");
                    socketChannel.configureBlocking(false);
                    //添加新的连接到selector种
                    socketChannel.register(selector, SelectionKey.OP_READ);
                }
                if (key.isReadable()) {
                    //读取数据
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    System.out.println("ready read message by : "+socketChannel.getRemoteAddress());
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024); //UTF-8种汉字占3个字节，所以这里最好是3的倍数，不然会出现乱码
                    String currentTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    if (readBuffer(readBuffer, socketChannel)) {
                        doWrite(socketChannel, currentTime);
                    }
                }
            }
        } catch (IOException e) {
            key.cancel();
            if (key.channel() != null) {
                key.channel().close();
            }
            e.printStackTrace();
        }
    }

    public void doWrite(SocketChannel channel, String response) throws IOException {
        if (response != null && !response.isEmpty()) {
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }

    private boolean readBuffer(ByteBuffer buffer, SocketChannel channel) throws IOException {
        int readBytes = channel.read(buffer);
        while (readBytes != -1 && readBytes != 0) {
            buffer.flip();
            while (buffer.hasRemaining()) {
                byte[] bytes = new byte[buffer.remaining()];
                buffer.get(bytes);
                String body = new String(bytes, StandardCharsets.UTF_8);
                if ("exit".equals(body)) {
                    doWrite(channel, "Bye!");
                    System.out.println(channel.socket().getRemoteSocketAddress() + " exit system");
                    channel.close();
                    return false;
                }
                System.out.print(body);
            }
            buffer.clear();
            readBytes = channel.read(buffer);
        }
        return true;
    }
}
